What is the "proper" way to ensure that all received (and currently active) messages are processed, before the application terminates?

I've implemented a simple SmartLifecycle which triggers the stop method on the listener containers, and then awaits ConsumerStoppingEvent. Do you see any issues with this solution?

@Slf4j
@Component
public class KafkaContainerListenerLifeCycle implements SmartLifecycle {

    private final Set<Integer> consumers = Collections.newSetFromMap(new ConcurrentHashMap<>());
    private final KafkaListenerEndpointRegistry registry;

    public KafkaContainerListenerLifeCycle(final KafkaListenerEndpointRegistry registry) {
        this.registry = registry;
    }

    @EventListener
    public void started(ConsumerStartedEvent event) {
        registry.getListenerContainers().forEach(c -> consumers.add(c.hashCode()));
    }

    @EventListener
    public void stopping(ConsumerStoppingEvent event) {
        registry.getListenerContainers().forEach(c -> consumers.remove(c.hashCode()));
    }

    @Override
    public void start() {

    }

    @Override
    public void stop() {
        registry.getListenerContainers().forEach(c -> c.stop());
    }

    @SneakyThrows
    @Override
    public void stop(Runnable callback) {
        stop();
        onShutdown();
        callback.run();
    }

    @Override
    public boolean isRunning() {
        return true;
    }

    private void onShutdown() throws InterruptedException {
        log.info("Shutting down the consumers");

        var numberOfActiveConsumers = consumers.size();

        log.info("Waiting on consumers to finish. {} active consumers", numberOfActiveConsumers);

        while (numberOfActiveConsumers > 0) {
            Thread.sleep(200L);

            numberOfActiveConsumers = consumers.size();
        }

        log.info("All consumers finished gracefully.");
    }

}


